package org.lagou.edu.serialization;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalPacket;
import com.alibaba.otter.canal.protocol.CanalPacket.Ack;
import com.alibaba.otter.canal.protocol.CanalPacket.Compression;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.google.common.base.Strings;
import com.google.protobuf.ByteString;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.serialization.ExtendedDeserializer;
import org.lagou.edu.serialization.dto.FlatMessage;
import org.lagou.edu.serialization.dto.OperationInfo;
import org.slf4j.MDC;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.util.CollectionUtils;

import java.util.*;

/**
 * @author shiding
 */
public class CanalMessageDeserializer implements ExtendedDeserializer<List<OperationInfo>> {

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {

    }

    @Override
    public List<OperationInfo> deserialize(String topic, byte[] data) {
        Message msgObj = deserializer(data);
        List<OperationInfo> opInfos = new ArrayList<>();
        if (msgObj != null) {
            System.out.println("Message:" + msgObj.getId() + ", 内容：" + JSON.toJSONString(msgObj));
            if (msgObj.getId() != -1 && msgObj.getEntries().size() > 0) {
                List<FlatMessage> flatMessages = this.messageConverter(msgObj);
                if (CollectionUtil.isNotEmpty(flatMessages)) {
                    for (FlatMessage flatMessage : flatMessages) {
                        List<OperationInfo> opInfoDTOs = recordToOpInfos(flatMessage);
                        if (!CollectionUtils.isEmpty(opInfoDTOs)) {
                            MDC.put("canal_transfer_time", String.valueOf(System.currentTimeMillis() - opInfoDTOs.get(0).getEs()));
                            System.out.println("内容：" + JSON.toJSONString(opInfoDTOs));
                            MDC.clear();

                            opInfos.addAll(opInfoDTOs);
                        }
                    }
                }
            }
        }

        return opInfos;
    }

    @Override
    public void close() {

    }

    /**
     * 将 byte[] 反序列化成Message
     *
     * @param data 数据
     * @return Message
     */
    private Message deserializer(byte[] data) {
        try {
            if (data == null) {
                return null;
            } else {
                CanalPacket.Packet p = CanalPacket.Packet.parseFrom(data);
                switch (p.getType()) {
                    case MESSAGES: {
                        if (!p.getCompression().equals(Compression.NONE)
                                && !p.getCompression().equals(Compression.COMPRESSIONCOMPATIBLEPROTO2)) {
                            throw new CanalClientException("compression is not supported in this connector");
                        }

                        CanalPacket.Messages messages = CanalPacket.Messages.parseFrom(p.getBody());
                        Message result = new Message(messages.getBatchId());

                        for (ByteString byteString : messages.getMessagesList()) {
                            result.addEntry(CanalEntry.Entry.parseFrom(byteString));
                        }
                        result.setRaw(false);

                        return result;
                    }
                    case ACK: {
                        Ack ack = Ack.parseFrom(p.getBody());
                        throw new CanalClientException("something goes wrong with reason: " + ack.getErrorMessage());
                    }
                    default: {
                        throw new CanalClientException("unexpected packet type: " + p.getType());
                    }
                }
            }
        } catch (Exception e) {
            throw new CanalClientException("deserializer failed", e);
        }
    }

    /**
     * 将Message转换为FlatMessage
     *
     * @param message 原生message
     * @return FlatMessage列表
     */
    private List<FlatMessage> messageConverter(Message message) {
        try {
            if (message == null) {
                return null;
            }

            List<FlatMessage> flatMessages = new ArrayList<>();
            List<CanalEntry.Entry> entrys;
            if (message.isRaw()) {
                List<ByteString> rawEntries = message.getRawEntries();
                entrys = new ArrayList<>(rawEntries.size());
                for (ByteString byteString : rawEntries) {
                    CanalEntry.Entry entry = CanalEntry.Entry.parseFrom(byteString);
                    entrys.add(entry);
                }
            } else {
                entrys = message.getEntries();
            }

            for (CanalEntry.Entry entry : entrys) {
                if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                        || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                    continue;
                }

                CanalEntry.RowChange rowChange;
                try {
                    rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:"
                            + entry.toString(), e);
                }

                CanalEntry.EventType eventType = rowChange.getEventType();

                FlatMessage flatMessage = new FlatMessage(message.getId());
                flatMessages.add(flatMessage);
                flatMessage.setPosition(entry.getHeader().getLogfileName() + "." + entry.getHeader().getLogfileOffset());
                flatMessage.setDatabase(entry.getHeader().getSchemaName());
                flatMessage.setTable(entry.getHeader().getTableName());
                flatMessage.setIsDdl(rowChange.getIsDdl());
                flatMessage.setType(eventType.toString());
                flatMessage.setEs(entry.getHeader().getExecuteTime());
                flatMessage.setTs(System.currentTimeMillis());
                flatMessage.setSql(rowChange.getSql());

                if (!rowChange.getIsDdl()) {
                    Map<String, Integer> sqlType = new LinkedHashMap<>();
                    Map<String, String> mysqlType = new LinkedHashMap<>();
                    List<Map<String, String>> data = new ArrayList<>();
                    List<Map<String, String>> old = new ArrayList<>();

                    Set<String> updateSet = new HashSet<>();
                    boolean hasInitPkNames = false;
                    for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                        if (eventType != CanalEntry.EventType.INSERT && eventType != CanalEntry.EventType.UPDATE
                                && eventType != CanalEntry.EventType.DELETE) {
                            continue;
                        }

                        Map<String, String> row = new LinkedHashMap<>();
                        List<CanalEntry.Column> columns;

                        if (eventType == CanalEntry.EventType.DELETE) {
                            columns = rowData.getBeforeColumnsList();
                        } else {
                            columns = rowData.getAfterColumnsList();
                        }

                        for (CanalEntry.Column column : columns) {
                            if (!hasInitPkNames && column.getIsKey()) {
                                flatMessage.addPkName(column.getName());
                            }
                            sqlType.put(column.getName(), column.getSqlType());
                            mysqlType.put(column.getName(), column.getMysqlType());
                            if (column.getIsNull()) {
                                row.put(column.getName(), null);
                            } else {
                                row.put(column.getName(), column.getValue());
                            }
                            // 获取update为true的字段
                            if (column.getUpdated()) {
                                updateSet.add(column.getName());
                            }
                        }

                        hasInitPkNames = true;
                        if (!row.isEmpty()) {
                            data.add(row);
                        }

                        if (eventType == CanalEntry.EventType.UPDATE) {
                            Map<String, String> rowOld = new LinkedHashMap<>();
                            for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
                                if (updateSet.contains(column.getName())) {
                                    if (column.getIsNull()) {
                                        rowOld.put(column.getName(), null);
                                    } else {
                                        rowOld.put(column.getName(), column.getValue());
                                    }
                                }
                            }
                            // update操作将记录修改前的值
                            if (!rowOld.isEmpty()) {
                                old.add(rowOld);
                            }
                        }
                    }
                    if (!sqlType.isEmpty()) {
                        flatMessage.setSqlType(sqlType);
                    }
                    if (!mysqlType.isEmpty()) {
                        flatMessage.setMysqlType(mysqlType);
                    }
                    if (!data.isEmpty()) {
                        flatMessage.setData(data);
                    }
                    if (!old.isEmpty()) {
                        flatMessage.setOld(old);
                    }
                }
            }
            return flatMessages;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * flatMessage 转化成OperationInfo
     *
     * @param flatMessage FlatMessage
     * @return OperationInfo
     */
    private List<OperationInfo> recordToOpInfos(FlatMessage flatMessage) {

        if (flatMessage != null && !CollectionUtils.isEmpty(flatMessage.getData())) {
            List<OperationInfo> opInfoDTOS = new ArrayList<>(flatMessage.getData().size());
            for (int i = 0; i < flatMessage.getData().size(); i++) {
                Map<String, String> data = flatMessage.getData().get(i);
                String pkName = CollectionUtils.isEmpty(flatMessage.getPkNames()) ? null : flatMessage.getPkNames().get(0);
                String pkId = null;
                if (!Strings.isNullOrEmpty(pkName)) {
                    pkId = data.get(pkName);
                }

                OperationInfo opInfoDTO;
                switch (flatMessage.getType()) {
                    case "INSERT":
                        opInfoDTO = new OperationInfo(flatMessage.getDatabase(), flatMessage.getTable(), pkId, flatMessage.getType(), data, flatMessage.getEs(), flatMessage.getTs());
                        opInfoDTO.setOldValue(null);
                        opInfoDTO.setId(flatMessage.getPosition());
                        opInfoDTOS.add(opInfoDTO);
                        break;

                    case "UPDATE":
                        opInfoDTO = new OperationInfo(flatMessage.getDatabase(), flatMessage.getTable(), pkId, flatMessage.getType(), data, flatMessage.getEs(), flatMessage.getTs());
                        Map<String, String> snapshot = new HashMap<>();
                        snapshot.putAll(data);
                        snapshot.putAll(flatMessage.getOld().get(i));
                        opInfoDTO.setOldValue(snapshot);
                        opInfoDTO.setId(flatMessage.getPosition());
                        opInfoDTOS.add(opInfoDTO);
                        break;

                    case "DELETE":
                        opInfoDTO = new OperationInfo(flatMessage.getDatabase(), flatMessage.getTable(), pkId, flatMessage.getType(), null, flatMessage.getEs(), flatMessage.getTs());
                        opInfoDTO.setOldValue(data);
                        opInfoDTO.setId(flatMessage.getPosition());
                        opInfoDTOS.add(opInfoDTO);
                        break;

                    default:
                        return null;
                }
            }
            return opInfoDTOS;
        }

        return null;
    }

    @Override
    public List<OperationInfo> deserialize(String topic, Headers headers, byte[] data) {
        Iterable<Header> headersIter = headers.headers(KafkaHeaders.RECEIVED_TIMESTAMP);
        while (headersIter.iterator().hasNext()) {
            MDC.put("kafka_transfer_time", new String(headersIter.iterator().next().value()));
        }

        List<OperationInfo> result = this.deserialize(topic, data);

        if (!Strings.isNullOrEmpty(MDC.get("kafka_transfer_time"))) {
            MDC.clear();
        }


        return result;
    }
}
